package com.chenxu.gmall.realtime.app.dwm;

import com.alibaba.fastjson.JSON;
import com.chenxu.gmall.realtime.bean.OrderWide;
import com.chenxu.gmall.realtime.bean.PaymentInfo;
import com.chenxu.gmall.realtime.bean.PaymentWide;
import com.chenxu.gmall.realtime.utils.DateTimeUtil;
import com.chenxu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * Date: 2021/07/17
 * Desc: 支付宽表处理程序
 * 从dwm_order_wide和dwd_payment_info读取数据，做数据合并；
 */
public class PaymentWideApp {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境的准备
        //1.1 创建流式处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        /*
        //1.3 检查点相关的配置
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/gmall/checkpoint/paymentWide"));
        */

        //TODO 2. 从kafka的主题中读取数据
        //2.1 声明相关的主题以及消费者组
        String paymentInfoSourceTopic = "dwd_payment_info";
        String orderWideSourceTopic = "dwm_order_wide";
        String paymentWideSinkTopic = "dwm_payment_wide";
        String groupId = "paymentwide_app_group";

        //2.1 读取支付数据
        FlinkKafkaConsumer<String> paymentInfoSource = MyKafkaUtil.getKafkaSource(paymentInfoSourceTopic, groupId);
        DataStreamSource<String> paymentInfoJsonStrDS = env.addSource(paymentInfoSource);

        //2.2 读取订单宽表数据
        FlinkKafkaConsumer<String> orderWideSource = MyKafkaUtil.getKafkaSource(orderWideSourceTopic, groupId);
        DataStreamSource<String> orderWideJsonStrDS = env.addSource(orderWideSource);

        //TODO 3. 对读取到的数据进行结构的转换   jsonStr->POJO
        //3.1 转换支付流
        SingleOutputStreamOperator<PaymentInfo> paymentInfoDS = paymentInfoJsonStrDS.map(
            jsonStr -> JSON.parseObject(jsonStr, PaymentInfo.class)
        );
        //3.2 转换订单流
        SingleOutputStreamOperator<OrderWide> orderWideDS = orderWideJsonStrDS.map(
            jsonStr -> JSON.parseObject(jsonStr, OrderWide.class)
        );

        //测试：//zk、kk、maxwell、hdfs、hbase、BaseDBApp、redis；
        //运行BaseDBApp；OrderWideApp；本应用；
        //运行rt_dblog下的业务数据生成脚本；
        //会输出pay和roderwidr两个数据流；
        //数据类型类似：
        //pay>>>>>:4> PaymentInfo{id=19914, order_id=30772, user_id=2312, total_amount=339.00,
        // subject='CAREMiLLE珂曼奶油小方口红 雾面滋润保湿持久丝缎唇膏 M02干玫瑰等6件商品', payment_type='1102',
        // create_time='2021-07-16 13:05:40', callback_time='2021-07-16 13:06:00'}

        //orderWide>>>>:4> OrderWide{detail_id=89344, order_id=30623, sku_id=17, order_price=6699.00, sku_num=1,
        // sku_name='TCL 65Q10 65英寸 QLED原色量子点电视 安桥音响 AI声控智慧屏 超薄全面屏 MEMC防抖 3+32GB 平板电视',
        // province_id=24, order_status='1001', user_id=178, total_amount=6717.00, activity_reduce_amount=0.00,
        // coupon_reduce_amount=0.00, original_total_amount=6699.00, feight_fee=18.00, split_feight_fee=null,
        // split_activity_amount=null, split_coupon_amount=null, split_total_amount=6699.00, expire_time='null',
        // create_time='2021-07-16 13:05:38', operate_time='null', create_date='null', create_hour='null',
        // province_name='湖北', province_area_code='420000', province_iso_code='CN-42', province_3166_2_code='CN-HB',
        // user_age=51, user_gender='F', spu_id=5, tm_id=4, category3_id=86,
        // spu_name='TCL巨幕私人影院电视 4K超高清 AI智慧屏  液晶平板电视机', tm_name='TCL', category3_name='平板电视'}

        //paymentInfoDS.print("pay>>>>>");
        //orderWideDS.print("orderWide>>>>");

        //TODO 4.设置Watermark以及提取事件时间字段
        //4.1 支付流的Watermark
        SingleOutputStreamOperator<PaymentInfo> paymentInfoWithWatermarkDS = paymentInfoDS.assignTimestampsAndWatermarks(
                //3秒的水位线设置；
                //由于下面用了interval join，所以这里3秒的水位线设置意义不大；
            WatermarkStrategy.<PaymentInfo>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<PaymentInfo>() {
                        @Override
                        public long extractTimestamp(PaymentInfo paymentInfo, long recordTimestamp) {
                            //需要将字符串的时间转换为毫秒数
                            return DateTimeUtil.toTs(paymentInfo.getCallback_time());
                        }
                    }
                )
        );
        //4.2 订单流的Watermark
        SingleOutputStreamOperator<OrderWide> orderWideWithWatermarkDS = orderWideDS.assignTimestampsAndWatermarks(
            WatermarkStrategy.<OrderWide>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<OrderWide>() {
                        @Override
                        public long extractTimestamp(OrderWide orderWide, long recordTimestamp) {
                            return DateTimeUtil.toTs(orderWide.getCreate_time());
                        }
                    }
                )
        );

        //TODO 5.对数据进行分组
        //5.1 支付流数据分组
        KeyedStream<PaymentInfo, Long> paymentInfoKeyedDS = paymentInfoWithWatermarkDS.keyBy(PaymentInfo::getOrder_id);
        //5.2 订单宽表流数据分组
        KeyedStream<OrderWide, Long> orderWideKeyedDS = orderWideWithWatermarkDS.keyBy(OrderWide::getOrder_id);

        //TODO 6.使用IntervalJoin关联两条流
        //订单信息join付款信息；
        //interval join的实现其实也是一种watermark的思想；两条流数据的当前时间中的较小者减去下边界就是watermark，小于watermark的就是过期数据，会被清除；
        SingleOutputStreamOperator<PaymentWide> paymentWideDS = paymentInfoKeyedDS
            .intervalJoin(orderWideKeyedDS)
                //考虑订单的预付款时间（半个小时时间）
            .between(Time.seconds(-1800), Time.seconds(0))
            .process(
                new ProcessJoinFunction<PaymentInfo, OrderWide, PaymentWide>() {
                    @Override
                    public void processElement(PaymentInfo paymentInfo, OrderWide orderWide, Context ctx, Collector<PaymentWide> out) throws Exception {
                        out.collect(new PaymentWide(paymentInfo, orderWide));
                    }
                }
            );

        paymentWideDS.print(">>>>");
        //TODO 7.将数据写到kafka的dwm层
        paymentWideDS.map(
            paymentWide->JSON.toJSONString(paymentWide)
        ).addSink(
            MyKafkaUtil.getKafkaSink(paymentWideSinkTopic)
        );

        //测试方法类似，查看一下dwm_payment_wide主题的消费情况；
        //数据结果为：
        //{"activity_reduce_amount":0.00,"callback_time":"2021-07-16 14:22:22","category3_id":61,"category3_name":"手机",
        // "coupon_reduce_amount":0.00,"detail_id":89977,"feight_fee":7.00,"order_create_time":"2021-07-16 14:22:01",
        // "order_id":31054,"order_price":1299.00,"order_status":"1001","original_total_amount":3726.00,
        // "payment_create_time":"2021-07-16 14:22:02","payment_id":20006,"payment_type":"1102",
        // "province_3166_2_code":"CN-GD","province_area_code":"440000","province_id":26,"province_iso_code":"CN-44",
        // "province_name":"广东","sku_id":6,"sku_name":"Redmi 10X 4G Helio G85游戏芯 4800万超清四摄 5020mAh大电量 小孔全面屏 128GB大存储 8GB+128GB 冰雾白 游戏智能手机 小米 红米",
        // "sku_num":1,"split_total_amount":1299.00,"spu_id":2,"spu_name":"Redmi 10X",
        // "subject":"索芙特i-Softto 口红不掉色唇膏保湿滋润 璀璨金钻哑光唇膏 Y01复古红 百搭气质 璀璨金钻哑光唇膏 等4件商品",
        // "tm_id":1,"tm_name":"Redmi","total_amount":3733.00,"user_age":24,"user_gender":"F","user_id":1431}
        env.execute();
    }
}
